package com.hzg.stream.join;

import com.hzg.bean.Record;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * @Author: huangzhigao
 * @Date: 2022/4/18 21:20
 */
public class JoinDemo {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        List<Record> records = new ArrayList<>();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        DataStreamSource<Record> source1 = env.fromCollection(records);
        DataStreamSource<Record> source2 = env.fromCollection(records);
        source1.join(source2).where(Record::getRecordTime) //first input key
                .equalTo(Record::getRecordTime)  //second input key
                .window(TumblingEventTimeWindows.of(Time.seconds(3), Time.seconds(2)))
                .apply(new JoinFunction<Record, Record, Object>() {
                    @Override
                    public Object join(Record first, Record second) throws Exception {
                        return null;
                    }
                });


        //intervalJoin  只能应用于eventTime
        KeyedStream<Record, Tuple> key1 = source1.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<Record>() {
            @Override
            public long extractAscendingTimestamp(Record element) {
                return element.getRecordTime();
            }
        }).keyBy(0);

        KeyedStream<Record, Tuple> key2 = source2.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<Record>() {
            @Override
            public long extractAscendingTimestamp(Record element) {
                return element.getRecordTime();
            }
        }).keyBy(0);

        key1.intervalJoin(key2).between(Time.seconds(0), Time.seconds(10)).process(new ProcessJoinFunction<Record, Record, Object>() {

            @Override
            public void processElement(Record left, Record right, Context ctx, Collector<Object> out) throws Exception {

            }
        });
    }
}
